package cn.jly.bigdata.spark.core

import org.apache.spark.rdd.RDD
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

/**
 * @author lanyangji
 * @date 2019/11/24 10:47
 */
object Spark02_expr13 {

  def main(args: Array[String]): Unit = {

    // local模式

    // 创建 SparkConf 对象
    // 这边也自定义了本地的模式（分配多少个cpu）
    // app id
    val sparkConf: SparkConf = new SparkConf().setMaster("local[*]").setAppName("wordCount")

    // 创建spark上下文对象，SparkContext -> sc
    val sc = new SparkContext(sparkConf)

    val arrayRDD: RDD[(Int, String)] = sc.makeRDD(Array((1, "aa"), (2, "bb"), (3, "cc"), (4, "dd")), 4)

    // partitionBy 分区算子
    // 传入分区器
    // 分区器中传入分区个数
    //val partitionByRDD: RDD[(Int, String)] = arrayRDD.partitionBy(new HashPartitioner(2))
    // 也可以传入自定义的分区器
    val partitionByRDD: RDD[(Int, String)] = arrayRDD.partitionBy(new MyPartitioner(2))

    println(partitionByRDD.getNumPartitions)

    partitionByRDD.saveAsTextFile("output")
//    for (elem <- partitionByRDD.glom().collect) {
//      println(elem.mkString(","))
//    }
  }
}

// 自定义分区器
// 继承spark提供的Partitioner抽象类
class MyPartitioner(partitions: Int) extends Partitioner {
  override def numPartitions: Int = partitions

  override def getPartition(key: Any): Int = {
    // 这边简单粗暴，全放在1号分区
    1
  }
}